package org.example.api.tableapi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.Csv;
import org.apache.flink.table.descriptors.Kafka;
import org.apache.flink.table.descriptors.Schema;

/**
 * @author huangqihan
 * @date 2021/3/8
 */
public class KafkaPipeLineTest {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 1
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2 连接 kafka，读取数据
        tableEnv.connect(
                new Kafka().version("0.11").topic("sensor")
                        .property("zookeeper.connect", "localhost:2181")
                        .property("bootstrap.servers", "localhost:9092")
        )
                .withFormat(new Csv())
                .withSchema(
                        new Schema()
                                .field("id", DataTypes.STRING())
                                .field("timestamp", DataTypes.BIGINT())
                                .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("inputTable");

        Table sensorTable = tableEnv.from("inputTable");

        // 简单查询
        Table resultTable = sensorTable.select("id,temp").filter("id === 'sensor_6'");

        // 聚合统计
        Table aggTable = sensorTable.groupBy("id").select("id, id.count as count, temp.avg as avgTemp");

        // 3 输出到kafka
        tableEnv.connect(
                new Kafka().version("0.11").topic("sink")
                        .property("zookeeper.connect", "localhost:2181")
                        .property("bootstrap.servers", "localhost:9092")
        )
                .withFormat(new Csv())
                .withSchema(
                        new Schema()
                                .field("id", DataTypes.STRING())
                                .field("timestamp", DataTypes.BIGINT())
                                .field("temp", DataTypes.DOUBLE())
                )
                .createTemporaryTable("outputTable");
        resultTable.insertInto("outputTable");

    }
}
